package com.cupdata.zicon.thrift.config;

import java.lang.reflect.Field;
import java.lang.reflect.UndeclaredThrowableException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.aopalliance.intercept.MethodInterceptor;
import org.aopalliance.intercept.MethodInvocation;
import org.apache.commons.pool2.KeyedObjectPool;
import org.apache.commons.pool2.KeyedPooledObjectFactory;
import org.apache.commons.pool2.impl.GenericKeyedObjectPoolConfig;
import org.apache.thrift.TException;
import org.apache.thrift.TServiceClient;
import org.apache.thrift.protocol.TProtocol;
import org.springframework.aop.framework.Advised;
import org.springframework.aop.framework.ProxyFactory;
import org.springframework.aop.support.AopUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.InvalidPropertyException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnWebApplication;
import org.springframework.cloud.client.ServiceInstance;
import org.springframework.cloud.client.discovery.DiscoveryClient;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.stereotype.Component;
import org.springframework.util.ReflectionUtils;
import org.springframework.util.StringUtils;

import com.cupdata.zicon.annotation.ThriftClient;
import com.cupdata.zicon.thrift.client.pool.ThriftClientKey;
import com.cupdata.zicon.thrift.client.pool.ThriftClientPool;
import com.cupdata.zicon.thrift.client.pool.ThriftClientPooledObjectFactory;

@Component
@Configuration
@ConditionalOnClass(ThriftClient.class)
@ConditionalOnWebApplication
public class ThriftClientAutoConfiguration implements org.springframework.beans.factory.config.BeanPostProcessor {

	@Value("${thirft.service.ids:''}")
	String thriftServerIds;
	
	@Value("${thrift.client.max.threads:8}")
	private int maxThreads;
	
	@Bean
	public KeyedObjectPool<ThriftClientKey, TServiceClient> thriftClientsPool() {
		if(StringUtils.isEmpty(thriftServerIds)){
			return null;
		}
		GenericKeyedObjectPoolConfig poolConfig = new GenericKeyedObjectPoolConfig();
		poolConfig.setMaxTotal(maxThreads);
		poolConfig.setMaxIdlePerKey(maxThreads);
		poolConfig.setMaxTotalPerKey(maxThreads);
		poolConfig.setJmxEnabled(false); // cause spring will autodetect itself
		return new ThriftClientPool(thriftClientPoolFactory(), poolConfig);
	}
	
	@Autowired
	private DiscoveryClient discoveryClient;

	@Autowired
	private KeyedObjectPool<ThriftClientKey, TServiceClient> thriftClientsPool;
	
	@Override
	public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
		return bean;
	}

	@Override
	public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
		if(StringUtils.isEmpty(thriftServerIds)){
			return bean;
		}
		Class clazz = bean.getClass();
		Object target = getTargetBean(bean);
		do {
			for (Field field : clazz.getDeclaredFields()) {
				if (field.isAnnotationPresent(ThriftClient.class)) {
					ThriftClient annotation = AnnotationUtils.getAnnotation(field, ThriftClient.class);
					ProxyFactory proxyFactory = getProxyFactoryForThriftClient(target, field);

					addPoolAdvice(proxyFactory, annotation);

					//proxyFactory.setFrozen(true);
					proxyFactory.setProxyTargetClass(true);

					ReflectionUtils.makeAccessible(field);
					ReflectionUtils.setField(field, target, proxyFactory.getProxy());

				}
			}
			clazz = clazz.getSuperclass();
		} while (clazz != null);
		return bean;
	}

	// We have to get a real bean in order to inject a thrift client into the
	// bean instead of its proxy.
	private Object getTargetBean(Object bean) {
		Object target = bean;
		while (AopUtils.isAopProxy(target)) {
			try {
				target = ((Advised) target).getTargetSource().getTarget();
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
		return target;
	}

	private ProxyFactory getProxyFactoryForThriftClient(Object bean, Field field) {
		ProxyFactory proxyFactory = null;
		try {
			proxyFactory = new ProxyFactory(
					BeanUtils.instantiateClass(field.getType().getConstructor(TProtocol.class), (TProtocol) null));
		} catch (NoSuchMethodException e) {
			throw new InvalidPropertyException(bean.getClass(), field.getName(), e.getMessage());
		}
		return proxyFactory;
	}

	private void addPoolAdvice(ProxyFactory proxyFactory, final ThriftClient annotation) {
		proxyFactory.addAdvice(new MethodInterceptor() {

			@Override
			public Object invoke(MethodInvocation invocation) throws Throwable {
				Object[] args = invocation.getArguments();

				Class<? extends TServiceClient> declaringClass = (Class<? extends TServiceClient>) invocation
						.getMethod().getDeclaringClass();

				TServiceClient thriftClient = null;

				ThriftClientKey key = new ThriftClientKey(declaringClass, annotation.serviceName(),
						annotation.serviceId());

				try {
					thriftClient = thriftClientsPool.borrowObject(key);
					return ReflectionUtils.invokeMethod(invocation.getMethod(), thriftClient, args);
				} catch (UndeclaredThrowableException e) {
					e.printStackTrace();
					if (TException.class.isAssignableFrom(e.getUndeclaredThrowable().getClass()))
						throw (TException) e.getUndeclaredThrowable();
					throw e;
				} finally {
					if (null != thriftClient)
						thriftClientsPool.returnObject(key, thriftClient);
				}
			}

		});
	}

	private KeyedPooledObjectFactory<ThriftClientKey, TServiceClient> thriftClientPoolFactory() {
		ThriftClientPooledObjectFactory pooledFactory = new ThriftClientPooledObjectFactory();
		ThriftClientPooledObjectFactory.setServiceMap(getServiceInfo());
		setDiscoveryClient();
		return pooledFactory;
	}

	public Map<String, List<ServiceInstance>> getServiceInfo() {
		String[] serviceIds = thriftServerIds.split(",");
		Map<String, List<ServiceInstance>> serviceMap = new HashMap<>();
		for (String serviceId : serviceIds) {
			List<ServiceInstance> intances = discoveryClient.getInstances(serviceId);
			serviceMap.put(serviceId, intances);
		}
		return serviceMap;
	}

	/**
	 * @param discoveryClient the discoveryClient to set
	 */
	public  void setDiscoveryClient() {
		ThriftClientPooledObjectFactory.setDiscoveryClient(discoveryClient);;
	}
	
	
	
	
}
